package com.horay.nettydemo.nio.socket;

import lombok.extern.slf4j.Slf4j;
import org.springframework.util.unit.DataSize;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.CharsetDecoder;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * Nio服务器
 *
 * @author heyingcheng
 * @date 2019/11/24 16:35
 */
@Slf4j
public class NioServer {

    private static final int SERVER_PORT = 6666;

    public static void main(String[] args) {
        try {
            ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
            InetSocketAddress inetSocketAddress = new InetSocketAddress(SERVER_PORT);
            serverSocketChannel.socket().bind(inetSocketAddress);
            serverSocketChannel.configureBlocking(false);

            Selector selector = Selector.open();

            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

            for (; ; ) {
                int op = selector.select(TimeUnit.SECONDS.toMillis(1L));
                if (op == 0) {
                    long l = System.currentTimeMillis();
                    if (l % 31 == 0) {
                        log.info("未监听到操作类型");
                    }
                    continue;
                }

                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                for (; iterator.hasNext(); ) {
                    SelectionKey selectionKey = iterator.next();
                    iterator.remove();

                    if (selectionKey.isAcceptable()) {
                        try {
                            ServerSocketChannel channel = (ServerSocketChannel) selectionKey.channel();
                            SocketChannel socketChannel = channel.accept();
                            socketChannel.configureBlocking(false);
                            socketChannel.register(selector, SelectionKey.OP_READ);
                            log.info("RemoteAddress:[{}], 注册客户端:[{}]", socketChannel.getRemoteAddress(), socketChannel.hashCode());
                        } catch (IOException e) {
                            log.error("接受客户端异常", e);
                        }
                    }

                    if (selectionKey.isReadable()) {
                        SocketChannel socketChannel = null;
                        try {
                            socketChannel = (SocketChannel) selectionKey.channel();
                            ByteBuffer byteBuffer = ByteBuffer.allocate(Math.toIntExact(DataSize.ofKilobytes(1).toBytes()));
                            CharsetDecoder decoder = StandardCharsets.UTF_8.newDecoder();
                            socketChannel.read(byteBuffer);

                            byteBuffer.flip();

                            String message = decoder.decode(byteBuffer).toString();
                            write(selector, socketChannel, message);
                            write(socketChannel, message);
                            log.info("客户端:[{}], 请求消息:[{}]", socketChannel.hashCode(), message);
                        } catch (IOException e) {
                            log.error("读取客户端消息异常", e);
                            if (Objects.nonNull(socketChannel)) {
                                socketChannel.close();
                            }
                            continue;
                        }
                    }

                    if (selectionKey.isWritable()) {
                        write(selectionKey);
                    }
                }
            }
        } catch (IOException e) {
            log.error("服务器启动异常", e);
        }

    }

    private static void write(Selector selector, SocketChannel selfSocketChannel, String message) {
        try {
            Set<SelectionKey> selectionKeys = selector.keys();
            Iterator<SelectionKey> iterator = selectionKeys.iterator();
            for (; iterator.hasNext(); ) {
                SelectionKey selectionKey = iterator.next();
                SelectableChannel channel = selectionKey.channel();
                if (channel instanceof SocketChannel && selfSocketChannel != channel) {
                    write((SocketChannel) channel, message);
                }
            }
        } catch (IOException e) {
            log.error("响应客户端消息异常", e);
        }
    }

    private static void write(SelectionKey selectionKey) {
        try {
            SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
            write(socketChannel, "服务器已接收消息");
        } catch (IOException e) {
            log.error("响应客户端消息异常", e);
        }
    }

    private static void write(SocketChannel socketChannel, String message) throws IOException {
        ByteBuffer byteBuffer = ByteBuffer.wrap(message.getBytes());
        socketChannel.write(byteBuffer);
        log.info("客户端:[{}], 响应消息:[{}]", socketChannel.hashCode(), new String(byteBuffer.array()));
    }

}
